package com.lry.netty01.server;

import com.lry.netty01.server.entity.AjaxResult;
import com.lry.netty01.server.entity.FixedFrame;
import com.lry.netty01.server.entity.Packet;
import com.lry.netty01.server.in.FrameDecoder;
import com.lry.netty01.server.in.HeartBeatHandler;
import com.lry.netty01.server.in.PacketDecoder;
import com.lry.netty01.server.in.PacketHandler;
import com.lry.netty01.server.out.PacketEncoder;
import com.lry.netty01.server.processor.HeartBeatRequestProcessor;
import com.lry.netty01.server.processor.RegisterRequestProcessor;
import com.lry.netty01.server.processor.RequestProcessor;
import com.lry.netty01.server.processor.WarningRequestProcessor;
import com.lry.netty01.server.queue.Task;
import com.lry.netty01.server.utils.ChannelUtil;
import com.lry.netty01.server.queue.TaskQueue;
import com.lry.netty01.server.utils.CommandUtil;
import com.lry.netty01.server.utils.Time;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.ClientCnxn;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.proto.ReplyHeader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class Server {

    private EventLoopGroup bossGroup,workerGroup,business;

    private Map<Integer,Channel> port_to_channel;

    private ServerBootstrap server;

    private static RequestProcessor firstRequestProcessor;

    @Autowired
    private TaskQueue taskQueue;

    private static Server tcpServer;

    @PostConstruct
    public void init() {
        tcpServer = this;
    }

    public void run(String[] args, Set<Integer> ports) throws InterruptedException {
        initEventLoopGroup();
        initBootstrap();
        initRequestProcessorChain();
        execTaskQueue();
        if(null!=ports&&!ports.isEmpty()){
            //监听端口
            startListen(ports);
        }
        //注册钩子
        shutDownHook();
    }

    private void initRequestProcessorChain() {

        WarningRequestProcessor warningRequestProcessor = new WarningRequestProcessor(null);

        HeartBeatRequestProcessor heartBeatRequestProcessor = new HeartBeatRequestProcessor(warningRequestProcessor);

        RegisterRequestProcessor registerRequestProcessor = new RegisterRequestProcessor(heartBeatRequestProcessor);

        firstRequestProcessor = registerRequestProcessor;
    }

    public static RequestProcessor getFirstRequestProcessor(){
        return firstRequestProcessor;
    }

    public static void offerTaskQueue(Task task){
        tcpServer.taskQueue.offerTask(task);
    }

    private void execTaskQueue() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                taskQueue.exec();
            }
        }).start();
    }

    private void initEventLoopGroup() {
        port_to_channel = new HashMap<>();

//        final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("tq-server-business-%d").setDaemon(false).build();
        // 接收客户端的TCP连接；向通信对端发送消息请求或者应答消息；
        bossGroup = new NioEventLoopGroup(1);
        //io线程
        workerGroup = new NioEventLoopGroup();
//        business = new NioEventLoopGroup(1000,threadFactory);
    }


    private void initBootstrap(){
        //服务器启动类，把所有组件串起来
        server = new ServerBootstrap();
        server.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                //.localAddress(new InetSocketAddress(port))
                .childOption(ChannelOption.SO_SNDBUF,1024)
                .childOption(ChannelOption.SO_RCVBUF,1024)
                //客户端连接同时涌入
                .option(ChannelOption.SO_BACKLOG,10000)
                //复用Time Wait 连接
                .childOption(ChannelOption.SO_REUSEADDR,true)
                //-1 表示close()方法立即返回，但OS底层会将发送缓冲区全部发送到对端。
                //.childOption(ChannelOption.SO_LINGER,-1)
                //低延迟，禁用nagle优化算法，禁止优化成大块数据传输
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.RCVBUF_ALLOCATOR,
                        new AdaptiveRecvByteBufAllocator(1, 1024, Integer.MAX_VALUE))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    //Handler主要是做编解码和业务逻辑处理
                    //而我们接收数据期望收到的可不是字节，而是基本数据类型或者对象。所以要做编解码
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new IdleStateHandler(600,10,0, TimeUnit.SECONDS));

                        //处理粘包，半包问题
                        ch.pipeline().addLast(new FrameDecoder(1024));

                        //将字节转成java对象吗，反序列化
                        ch.pipeline().addLast("decoder", new PacketDecoder());

                        //将java对象转换字节，序列化
                        ch.pipeline().addLast("encoder",new PacketEncoder());

                        //心跳处理器
                        ch.pipeline().addLast("heartBeat",new HeartBeatHandler());

                        //报文处理器
                        ch.pipeline().addLast("packetHandler",new PacketHandler());
                    }
                });
    }


    private void startListen(Set<Integer> ports) throws InterruptedException {
        for(Integer port:ports){
            addPort(port);
        }
    }

    //像addPort和delPort可以提供给web端调用，动态改变监听端口
    public AjaxResult addPort(Integer port)  {
        if(null==port){
            return AjaxResult.error("参数有误，端口为空");
        }
        Set<Integer> ports = port_to_channel.keySet();
        if(ports.contains(port)){
            return AjaxResult.error(ports+"中："+ port+"已存在");
        }
        //绑定端口
        ChannelFuture future = null;
        try {
            future = server.bind(port).sync();
        } catch (InterruptedException e) {
            log.info("bind 端口失败"+port);
            e.printStackTrace();
            return AjaxResult.error("bind 端口失败"+port);
        }
        if(future.isSuccess()){
            log.info("Netty server listening on port "+port+" and ready for connections...");
            return AjaxResult.success("bind 端口成功"+port);
        }
        return AjaxResult.error("bind 端口future失败"+port);
    }

    public AjaxResult delPort(Integer port){
        if(null==port){
            return AjaxResult.error("参数有误，端口为空");
        }
        Set<Integer> ports = port_to_channel.keySet();
        if(!ports.contains(port)){
            return AjaxResult.error(ports+"中："+ port+"不存在");
        }
        Channel channel = port_to_channel.get(port);
        ChannelFuture future = channel.closeFuture().syncUninterruptibly();
        if(future.isSuccess()){
            port_to_channel.remove(port);
            log.info("netty stop listen on port "+ port);
            log.info("After stop -----Netty server listening on port "+port_to_channel.keySet()+" and ready for connections...");
            return AjaxResult.success("解绑成功");
        }
        return AjaxResult.error("解绑失败");
    }


    private void shutDownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(()-> {
            log.info("服务器关闭，开始释放资源，记录设备下线时间");
            releaseResource();
        }));
    }

    private void releaseResource() {
        //关闭所有通道
        Set<Channel> channelSet = ChannelUtil.online_channels.keySet();
        for(Channel channel:channelSet){
            if(ChannelUtil.checkChannel(channel)){
                channel.close();
            }
        }

        try {
            if(bossGroup!=null) {
                bossGroup.shutdownGracefully().sync();
            }
            if(workerGroup!=null) {
                workerGroup.shutdownGracefully().sync();
            }
            if(business!=null) {
                business.shutdownGracefully().sync();
            }
        } catch (InterruptedException e) {
            log.info("bossGroup workerGroup business shutdown 异常");
            e.printStackTrace();
        }
    }



}
